package com.yc.rpc.core.server;

import com.alibaba.fastjson.JSONObject;
import com.yc.rpc.core.Metadata;
import com.yc.rpc.core.RpcClientConfig;
import com.yc.rpc.core.RpcServerConfig;
import com.yc.rpc.core.client.NettyClientHandler;
import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.Unpooled;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.LineBasedFrameDecoder;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.TimeUnit;


public class HundunClient {

    private volatile Channel channel;

    RpcServerConfig rpcServerConfig;

    public HundunClient(RpcServerConfig rpcServerConfig) {
        this.rpcServerConfig = rpcServerConfig;
    }

    @Autowired
    NettyClientHandler nettyClientHandler;


    public void connect(String host, int port) {
        EventLoopGroup worker = new NioEventLoopGroup();
        try {
            Bootstrap b = new Bootstrap();
            /**
             *EventLoop的组
             */
            b.group(worker);
            /**
             * 用于构造socketchannel工厂
             */
            b.channel(NioSocketChannel.class);
            /**设置选项
             * 参数：Socket的标准参数（key，value），可自行百度
             保持呼吸，不要断气！
             * */
            b.option(ChannelOption.SO_KEEPALIVE, true);
            /**
             * 自定义客户端Handle（客户端在这里搞事情）
             */
            b.handler(new ChannelInitializer<SocketChannel>() {
                @Override
                public void initChannel(SocketChannel ch){
                    ch.pipeline().addLast(new LineBasedFrameDecoder(Integer.MAX_VALUE));
                    ch.pipeline().addLast(new HundunHandler());
//                    ch.pipeline().addLast(new HundunHandler());
                }
            });


            b.connect(host,port).addListener((ChannelFutureListener) future -> {
                // 连接失败
                if (!future.isSuccess()) {
                    System.out.printf("[start][Netty Client 连接服务器(%s:%s) 失败]\n", host, port);
                    reconnect(host,port);
                    return;
                }
                // 连接成功
                channel = future.channel();
                System.out.printf("[start][Netty Client 连接服务器(%s:%s) 成功]\n", host, port);
                new Thread(()->{
                    for (;;) {

                        ServerDomain serverDomain = new ServerDomain();
                        serverDomain.setAddress(rpcServerConfig.getHost() + ":" + rpcServerConfig.getPort());
                        serverDomain.setServerId(rpcServerConfig.getServerId());

                        ServerCommand serverCommand = new ServerCommand();
                        serverCommand.setServerDomain(serverDomain);
                        serverCommand.setServerType("REGISTER");

                        channel.writeAndFlush(Unpooled.buffer().writeBytes((JSONObject.toJSONString(serverCommand)+ "\r\n").getBytes()));

                    ServerCommand getServerCommand = new ServerCommand();
                    getServerCommand.setServerType("GET");

                    channel.writeAndFlush(Unpooled.buffer().writeBytes((JSONObject.toJSONString(getServerCommand)+ "\r\n").getBytes()));

                        try {
                            Thread.sleep(5000);
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                    }
                }).start();
                });



        } finally {

//            worker.shutdownGracefully();
        }
    }

    public void reconnect(String host, int port) {
        connect(host,port);
    }


    public SynchronousQueue<String> send(Metadata msg) {
        return nettyClientHandler.send(channel, msg);
    }



//
//    public Channel doConnect(SocketAddress address) throws InterruptedException {
//        ChannelFuture future = worker.connect(address);
//        Channel channel = future.sync().channel();
//        return channel;
//    }


//    public static void main(String[] args) throws Exception {
//        NettyClient client=new NettyClient();
//        client.connect("127.0.0.1", 8088).sync();
//
//    }
}